package hyl.base.kv;

//import static org.fusesource.leveldbjni.JniDBFactory.*;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBException;
import org.iq80.leveldb.DBFactory;
import org.iq80.leveldb.DBIterator;
import org.iq80.leveldb.Options;
import org.iq80.leveldb.ReadOptions;
import org.iq80.leveldb.Snapshot;
import org.iq80.leveldb.WriteBatch;
//import org.iq80.leveldb.impl.Iq80DBFactory;
import org.iq80.leveldb.impl.Iq80DBFactory;

import hyl.core.MyFun;
import hyl.core.fun.MyByte;
import hyl.core.info.MyKV;
import hyl.core.io.MyFile;
import hyl.core.reflect.MyKryo;
import hyl.core.run.IDo1;
import hyl.core.run.IDoEachT;
import hyl.core.run.IDoMapT;
import hyl.core.run.IFilter;

/**
 * 键的存储控制器 封装的是 rocksdb
 * 
 * @author 37798955@qq.com
 *
 */
public class KVDb {
	MyKryo kryo = MyKryo.getInstance();
	DB _db;
	public int I模式 = KVConfigs.I模式_单机;

	public String s参数 = null;

	public String getS参数() {
		return s参数;
	}

	public void setS参数(String s参数) {
		this.s参数 = s参数;
	}

	public int getI模式() {
		return I模式;
	}

	public void setI模式(int i模式) {
		I模式 = i模式;
	}

	void 备份(byte[] key, byte[] value) {
		if (I模式 == KVConfigs.I模式_单机)
			return;

	}

	public <T> byte[] do序列化(T 对象) {
		return kryo.serialize(对象);
	}

	public <T> T do反序列化(byte[] 对象数据) {
		return kryo.deserialize(对象数据);
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @return
	 */
	DBIterator getIterator() {
		Snapshot snapshot = _db.getSnapshot();
		// 读选项
		ReadOptions readOptions = new ReadOptions();
		readOptions.fillCache(false);// 遍历中swap出来的数据，不应该保存在memtable中。
		readOptions.snapshot(snapshot);// 默认snapshot为当前。
		DBIterator iterator = _db.iterator(readOptions);

		return iterator;

	}

	DBIterator getIterator(byte[] 前缀) {
		DBIterator iterator = _db.iterator();// getIterator();//
		if (前缀 == null)
			iterator.seekToFirst();
		else
			iterator.seek(前缀);
		return iterator;
	}

	/**
	 * 缓存所有正在用的kvdb文件
	 */
	// final static Map<String, DB> _dbs = new HashMap<>();
	final static Map<String, KVDb> _dbs = new HashMap<>();
	public static KVDb getInstance(String 主目录, int 缓存大小) {
		return getInstance(主目录, null, KVConfigs.ini缓存大小(缓存大小));
	}
	public static KVDb getInstance(String 主目录, Options 选项) {
		return getInstance(主目录, null, 选项);
	}

	public static KVDb getInstance(String 主目录, String 类路径, Options 选项) {
		if (!MyFun.isEmpty(类路径)) {
			主目录 = 主目录 + File.separator + 类路径;
		}
		KVDb db = _dbs.get(主目录);
		if (db == null) {
			db = new KVDb(主目录, 选项);
			_dbs.put(主目录, db);
		}
		return db;
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param options
	 * @param path
	 * @throws IOException
	 */

	/**
	 * 在配置路径下添加子目录
	 * 
	 * @param 主目录
	 * @param 类路径
	 * @param 选项
	 */
	private KVDb(String 主目录, Options 选项) {
		File file = MyFile.openDir(主目录);
		DBFactory factory = new Iq80DBFactory();
		try {
			_db = factory.open(file, 选项);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void close() {
		try {
			_db.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public String get数据库状态() {
		final String str = _db.getProperty("leveldb.stats");
		return str;
	}

	/**
	 * 从保存的key中 获取排除前缀后剩余的后缀
	 * 
	 * @param key
	 * @param 前缀
	 * @return
	 */
	public static byte[] sub后缀(byte[] key, String 前缀) {
		return MyByte.subBytes(key, 前缀.getBytes().length);
	}

	/**
	 * 从保存的key中 获取排除前缀后剩余的后缀
	 * 
	 * @param key
	 * @param 前缀
	 * @return
	 */
	public static byte[] sub后缀(byte[] key, byte[] 前缀) {
		return MyByte.subBytes(key, 前缀.length);
	}

	/**
	 * 已测
	 * 
	 * @param 路径
	 * @param key
	 * @return
	 */
	public static byte[] concat(String 路径, String key) {
		byte[] pathkey;
		if (MyFun.isEmpty(路径))
			pathkey = MyFun.u8str2Bytes(key);
		else if (MyFun.isEmpty(key))
			pathkey = MyFun.u8str2Bytes(路径);
		else
			pathkey = MyFun.concat(MyFun.u8str2Bytes(路径), MyFun.u8str2Bytes(key));
		return pathkey;
	}

	/**
	 * 已测
	 * 
	 * @param 路径
	 * @param key
	 * @return
	 */
	public static byte[] concat(String 路径, byte[] key) {
		byte[] pathkey;
		if (MyFun.isEmpty(路径))
			pathkey = key;
		else if (MyFun.isEmpty(key))
			pathkey = MyFun.u8str2Bytes(路径);
		else
			pathkey = MyFun.concat(MyFun.u8str2Bytes(路径), key);
		return pathkey;
	}

	/**
	 * 已测
	 * 
	 * @param 路径
	 * @param key
	 * @return
	 */
	public static byte[] concat(byte[] 路径, byte[] key) {
		byte[] pathkey;
		if (MyFun.isEmpty(路径))
			pathkey = key;
		else if (MyFun.isEmpty(key))
			pathkey = 路径;
		else
			pathkey = MyFun.concat(路径, key);
		return pathkey;
	}

	/**
	 * 已测
	 * 
	 * @param 路径
	 * @param key
	 * @return
	 */
	public static byte[] concat(byte[] 路径, String key) {
		byte[] pathkey;
		if (MyFun.isEmpty(路径))
			pathkey = MyFun.u8str2Bytes(key);
		else if (MyFun.isEmpty(key))
			pathkey = 路径;
		else
			pathkey = MyFun.concat(路径, MyFun.u8str2Bytes(key));
		return pathkey;
	}

	//////////////////////////////// 路径+key+value模型/////////////////////////////////////////
	/**
	 * 插入一条数据 没有路径,key就包含了路径 <br>
	 * 已测
	 * 
	 * @param key   byte类型
	 * @param value
	 */
	public boolean putB(byte[] key, byte[] value) {
		try {
			_db.put(key, value);
		} catch (DBException e) {
			return false;
		}
		return true;
	}

	/**
	 * 插入一条数据 <br>
	 * 已测
	 * 
	 * @param key   字符串类型
	 * @param value
	 */
	public boolean putS(String key, byte[] value) {
		return putB(MyFun.u8str2Bytes(key), value);
	}

	/**
	 * 插入一条数据 <br>
	 * 已测
	 * 
	 * @param key   long 类型
	 * @param value
	 */
	public boolean putL(long key, byte[] value) {
		return putB(MyFun.long2ByteArray(key), value);
	}

	/**
	 * 插入一条数据 <br>
	 * 已测
	 * 
	 * @param key   int 类型
	 * @param value
	 */
	public boolean putI(int key, byte[] value) {
		return putB(MyFun.int2ByteArray(key), value);
	}

	public boolean putT(byte[] value) {
		return putB(MyFun.getNsBytes(), value);
	}

	/**
	 * 插入一条数据 key=路径+后缀 <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀    字符串类型
	 * @param value
	 */
	public boolean putSs(String 路径, String 后缀, byte[] value) {
		return putB(concat(路径, 后缀), value);
	}

	/**
	 * 插入一条数据 key=路径+后缀 <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀    是byte[] 类型
	 * @param value
	 */
	public boolean putSb(String 路径, byte[] 后缀, byte[] value) {
		return putB(concat(路径, 后缀), value);

	}

	/**
	 * 插入一条数据 key=主题+时间戳 <br>
	 * 已测
	 * 
	 * @param 主题
	 * @param value
	 */
	public boolean putSt(String 主题, byte[] value) {
		byte[] 前缀 = MyFun.u8str2Bytes(主题);
		return putBt(前缀, value);
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀
	 * @param value
	 * @return
	 */
	public boolean putBs(byte[] 路径, String 后缀, byte[] value) {
		return putB(concat(路径, 后缀), value);
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀
	 * @param value
	 * @return
	 */
	public boolean putBb(byte[] 路径, byte[] 后缀, byte[] value) {
		return putB(concat(路径, 后缀), value);

	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param value
	 * @return
	 */

	public boolean putBt(byte[] 前缀, byte[] value) {
		byte[] key = MyFun.concat(前缀, MyFun.getNsBytes());
		return putB(key, value);
	}

//////////////////////////////////del////////////////////////
	/**
	 * 已测
	 * 
	 * @param 主题
	 * @param key
	 * @return
	 */
	public boolean delB(byte[] key) {
		try {
			_db.delete(key);
			return true;
		} catch (DBException e) {
			return false;
		}
	}

	/**
	 * 已测
	 * 
	 * @param 主题
	 * @param key
	 * @return
	 */
	public boolean delS(String key) {
		return delB(MyFun.u8str2Bytes(key));

	}

	/**
	 * 已测
	 * 
	 * @param 主题
	 * @param key
	 * @return
	 */
	public boolean delSb(String 主题, byte[] key) {
		return delB(concat(主题, key));
	}

	/**
	 * 已测
	 * 
	 * @param 主题
	 * @param key
	 * @return
	 */
	public boolean delBs(byte[] 主题, String key) {
		return delB(concat(主题, key));
	}

	/**
	 * 已测
	 * 
	 * @param 主题
	 * @param key
	 * @return
	 */
	public boolean delBb(byte[] 主题, byte[] key) {
		return delB(concat(主题, key));
	}

	/**
	 * 已测
	 * 
	 * @param 主题
	 * @param key
	 * @return
	 */
	public boolean delSs(String 主题, String key) {
		return delB(concat(主题, key));
	}

	/**
	 * 已测
	 * 
	 * @return
	 */
	public boolean clear() {
		return clear(null);
	}

	/**
	 * 已测
	 * 
	 * @param 主题
	 * @return
	 */
	public boolean clear(byte[] 主题) {

		DBIterator iterator = getIterator(主题);
		WriteBatch batch = _db.createWriteBatch();
		boolean flag = true;
		try {
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();
				// MyFun.print("clear--", MyFun.bytes2U8str(mp.getKey()),
				// MyFun.bytes2U8str(主题));
				if (主题 == null || 主题.length == 0 || MyByte.leftLike(mp.getKey(), 主题))
					batch.delete(mp.getKey());
			}
			_db.write(batch);
		} catch (Exception e) {
			flag = false;
		} finally {
			try {
				iterator.close();
				batch.close();
			} catch (IOException e) {
				e.printStackTrace();
				flag = false;
			}
		}
		return flag;
	}

	/////////////////////////////////////// 以上是公共部分////////////////////////////////////
	/**
	 * <br>
	 * 测过
	 * 
	 * @param key
	 * @return
	 */
	public byte[] getB(byte[] key) {
		return _db.get(key);
	}

	/**
	 * 获取键值对 <br>
	 * 测过
	 * 
	 * @param key 字符串类型
	 * @return byte[] 类型
	 */
	public byte[] getS(String key) {
		return getB(MyFun.u8str2Bytes(key));
	}

	/**
	 * <br>
	 * 测过
	 * 
	 * @param key
	 * @return
	 */
	public byte[] getL(long key) {
		return getB(MyFun.long2ByteArray(key));
	}

	/**
	 * 如果keys 很多 速度慢<br>
	 * 测过
	 * 
	 * @param keys
	 * @return
	 */
	public Map<byte[], byte[]> getManyB(List<byte[]> keys) {
		HashMap<byte[], byte[]> map = new HashMap<>();
		for (byte[] key : keys) {
			map.put(key, _db.get(key));
		}
		return map;

	}

	/**
	 * 如果keys 很多 速度慢<br>
	 * 测过
	 * 
	 * @param keys
	 * @return
	 */
	public Map<String, byte[]> getManyS(List<String> keys) {
		Map<String, byte[]> map = new HashMap<>();
		for (String key : keys) {
			byte[] kb = MyFun.u8str2Bytes(key);
			map.put(key, _db.get(kb));
		}
		return map;
	}

	/**
	 * 获取多条记录结果<br>
	 * 测过
	 * 
	 * 
	 * @param 前缀
	 * @param keys
	 * @return
	 */
	public Map<byte[], byte[]> getManyBb(byte[] 前缀, List<byte[]> keys) {
		HashMap<byte[], byte[]> map = new HashMap<>();
		for (byte[] key : keys) {
			byte[] k = concat(前缀, key);
			map.put(k, _db.get(k));
		}
		return map;
	}

	/**
	 * 取n条记录
	 * 
	 * @param 前缀
	 * @param 取n条记录
	 * @return
	 */
	public Map<byte[], byte[]> getRangeB(byte[] 前缀, int 取n条记录) {
		HashMap<byte[], byte[]> map = new HashMap<>();
		IDoEachT<byte[], byte[]> call = new IDoEachT<byte[], byte[]>() {
			int i = 0;

			@Override
			public boolean isOver(byte[] key, byte[] val) throws Exception {
				map.put(key, val);
				if ((++i) >= 取n条记录)
					return true;// 停止取数
				else
					return false;
			}
		};
		do遍历操作(前缀, call);
		return map;
	}

	/**
	 * 字节数据 取出后直接反序列化成对象<br>
	 * 避免再次遍历 <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @return
	 */
	public <T> List<T> getLikeBo(byte[] 前缀) {
		DBIterator iterator = getIterator(前缀);
		List<T> list = new ArrayList<>();
		while (iterator.hasNext()) {
			Map.Entry<byte[], byte[]> mp = iterator.next();
			if (MyByte.leftLike(mp.getKey(), 前缀))
				list.add(kryo.deserialize(mp.getValue()));
		}
		return list;
	}

	/**
	 * 取出数据列表 但是不序列化 <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @return
	 */
	public List<byte[]> values(byte[] 前缀) {
		DBIterator iterator = getIterator(前缀);
		List<byte[]> list = new ArrayList<>();
		while (iterator.hasNext()) {
			Map.Entry<byte[], byte[]> mp = iterator.next();
			if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀))
				list.add(mp.getValue());
		}
		return list;
	}

	/**
	 * 已测
	 * 
	 * @return
	 */
	public List<byte[]> values() {
		return values(null);
	}

	/**
	 * 已测
	 * 
	 * @return
	 */
	public Map<byte[], byte[]> getAll() {
		Map<byte[], byte[]> map = new HashMap<>();
		IDoEachT<byte[], byte[]> call = new IDoEachT<byte[], byte[]>() {
			@Override
			public boolean isOver(byte[] key, byte[] val) throws Exception {
				map.put(key, val);
				return false;
			}
		};
		do遍历操作(null, call);
		return map;
	}

	public Map<String, String> getAllS() {
		Map<String, String> map = new HashMap<>();
		IDoEachT<byte[], byte[]> call = new IDoEachT<byte[], byte[]>() {
			@Override
			public boolean isOver(byte[] key, byte[] val) throws Exception {
				map.put(MyFun.bytes2U8str(key), MyFun.bytes2U8str(val));
				return false;
			}
		};
		do遍历操作(null, call);
		return map;
	}

	public List<byte[]> keys() {
		return keys(null);
	}

	public List<byte[]> keys(byte[] 前缀) {
		DBIterator iterator = getIterator(前缀);
		List<byte[]> list = new ArrayList<>();
		while (iterator.hasNext()) {
			Map.Entry<byte[], byte[]> mp = iterator.next();
			if (MyByte.leftLike(mp.getKey(), 前缀))
				list.add(mp.getKey());
		}
		return list;
	}

	//////////////////////////////// 路径+key+value模型/////////////////////////////////////////

	/**
	 * 返回 数据 key =路径+后缀<br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀
	 * @return
	 */
	public byte[] getSs(String 路径, String 后缀) {
		return getB(concat(路径, 后缀));
	}

	public byte[] getBs(byte[] 路径, String 后缀) {
		return getB(concat(路径, 后缀));
	}

	/**
	 * 返回 数据 key =路径+后缀<br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀
	 * @return
	 */
	public byte[] getSb(String 路径, byte[] 后缀) {
		return getB(concat(路径, 后缀));
	}

	/**
	 * 返回 数据 key =路径+后缀<br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 后缀
	 * @return
	 */
	public byte[] getBb(byte[] 路径, byte[] 后缀) {
		return getB(concat(路径, 后缀));
	}

/////////////////////////////////put//////////////////////////////////////////////////
	/**
	 * 插入多条数据 <br>
	 * 已测
	 * 
	 * @param 批量数据
	 */
	public boolean putManyB(Map<byte[], byte[]> 批量数据) {
		return do批量操作((batch) -> {
			for (Entry<byte[], byte[]> pp : 批量数据.entrySet()) {
				batch.put(pp.getKey(), pp.getValue());
			}
		});
	}

	/**
	 * 插入多条数据 <br>
	 * 已测
	 * 
	 * @param 批量数据
	 */
	public boolean putManyB(List<byte[]> 批量数据) {
		return do批量操作((batch) -> {
			for (byte[] pp : 批量数据) {
				batch.put(MyFun.getNsBytes(), pp);
			}
		});
	}

	/**
	 * 插入多条数据<br>
	 * 已测
	 * 
	 * @param 批量数据
	 */
	public boolean putManyS(Map<String, byte[]> 批量数据) {
		return do批量操作((batch) -> {
			for (Entry<String, byte[]> pp : 批量数据.entrySet()) {
				batch.put(MyFun.u8str2Bytes(pp.getKey()), pp.getValue());
			}
		});
	}

	/**
	 * 已测<br>
	 * 
	 * @param 批量数据
	 * @return
	 */
	public boolean putManyS(List<String> 批量数据) {
		return do批量操作((batch) -> {
			for (String pp : 批量数据) {
				batch.put(MyFun.getNsBytes(), MyFun.u8str2Bytes(pp));
			}
		});
	}

	/**
	 * 已测<br>
	 * 
	 * @param 路径
	 * @param 批量数据
	 * @return
	 */
	public <T> boolean putManyBo(String 路径, List<T> 批量数据) {
		byte[] path = MyFun.u8str2Bytes(路径);
		return putManyBo(path, 批量数据);
	}

	/**
	 * 已测<br>
	 * 
	 * @param 路径
	 * @param 批量数据
	 * @return
	 */
	public <T> boolean putManyBo(byte[] 路径, List<T> 批量数据) {
		return do批量操作((batch) -> {
			for (T pp : 批量数据) {
				byte[] b = kryo.serialize(pp);
				byte[] pathkey = concat(路径, MyFun.int2ByteArray(pp.hashCode()));
				batch.put(pathkey, b);
			}
		});
	}

	/**
	 * 已测<br>
	 * 
	 * @param 路径
	 * @param 批量数据 插入的时候补充时间戳
	 */
	public boolean putManySt(String 路径, List<byte[]> 批量数据) {
		return putManyBt(MyFun.u8str2Bytes(路径), 批量数据);
	}

	/**
	 * 已测<br>
	 * 
	 * @param 路径
	 * @param 批量数据
	 * @return
	 */
	public boolean putManyBt(byte[] 路径, List<byte[]> 批量数据) {
		return do批量操作((batch) -> {
			for (byte[] pp : 批量数据) {
				byte[] pathkey = concat(路径, MyFun.getNsBytes());
				batch.put(pathkey, pp);
			}
		});
	}

	/**
	 *
	 * 插入多条数据 不适合多线程使用 <br>
	 * 已测
	 * 
	 * @param 开始
	 * @param 批量数据
	 */
	public boolean putManyBn(byte[] 路径, long 开始, List<byte[]> 批量数据) {
		WriteBatch batch = _db.createWriteBatch();
		try {
			for (byte[] pp : 批量数据) {
				byte[] pathkey = concat(路径, MyFun.long2ByteArray(开始));
				batch.put(pathkey, pp);
				开始++;
			}
			_db.write(batch);
		} finally {
			try {
				batch.close();
			} catch (IOException e) {
				e.printStackTrace();
				return false;
			}
		}
		return true;

	}

	/**
	 * 
	 * 批量插入多条数据 key=路径+批量数据.key <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 批量数据
	 */
	public boolean putManySb(String 路径, Map<byte[], byte[]> 批量数据) {
		return putManyBb(MyFun.u8str2Bytes(路径), 批量数据);
	}

	/**
	 * 批量插入多条数据 key=路径+批量数据.key <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 批量数据
	 * @return
	 */
	public boolean putManyBb(byte[] 路径, Map<byte[], byte[]> 批量数据) {
		return do批量操作((batch) -> {
			for (Entry<byte[], byte[]> pp : 批量数据.entrySet()) {

				byte[] pathkey = concat(路径, pp.getKey());
				batch.put(pathkey, pp.getValue());
			}
		});
	}

	/**
	 * 
	 * 批量插入多条数据 key=路径+批量数据.key<br>
	 * 把key 分解成 路径+后缀模式 <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 批量数据 <后缀,数据>
	 */
	public boolean putManySs(String 路径, Map<String, byte[]> 批量数据) {
		return putManyBs(MyFun.u8str2Bytes(路径), 批量数据);
	}

	/**
	 * 批量插入多条数据 key=路径+批量数据.key<br>
	 * <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 批量数据
	 * @return
	 */
	public boolean putManyBs(byte[] 路径, Map<String, byte[]> 批量数据) {
		return do批量操作((batch) -> {
			for (Entry<String, byte[]> pp : 批量数据.entrySet()) {
				byte[] pathkey = concat(路径, pp.getKey());
				batch.put(pathkey, pp.getValue());
			}
		});
	}

	/**
	 * 
	 * 批量插入多条数据 key=路径+批量数据.key<br>
	 * 把key 分解成 路径+时间戳模式 <br>
	 * 已测
	 * 
	 * @param 路径
	 * @param 批量数据 <时间戳,数据>
	 */
	public boolean putManySi(String 路径, Map<Integer, byte[]> 批量数据) {

		return do批量操作((batch) -> {
			for (Entry<Integer, byte[]> pp : 批量数据.entrySet()) {
				byte[] pathkey = concat(路径, MyFun.int2ByteArray(pp.getKey()));
				batch.put(pathkey, pp.getValue());
			}

		});

	}

	/////////////////////////////// like////////////////////////
	public void likeBb(byte[] 前缀, byte[] 键的关键字, IDoEachT<byte[], byte[]> call) {
		if (call == null)
			return;
		int n = MyFun.get字节数(前缀);
		do遍历操作(前缀, (k, i) -> {
			// ByteUtils.subBytes(k, i)
			// 不包含键的关键字就继续遍历,如果包含 运行回调函数 call
			if (MyByte.like(k, 键的关键字, n))
				return call.isOver(k, i);
			return false;
		});
	}

	public void likeSb(String 前缀, byte[] 键的关键字, IDoEachT<byte[], byte[]> call) {
		likeBb(MyFun.u8str2Bytes(前缀), 键的关键字, call);
	}

	public void likeSs(String 前缀, String 键的关键字, IDoEachT<byte[], byte[]> call) {
		likeBb(MyFun.u8str2Bytes(前缀), MyFun.u8str2Bytes(键的关键字), call);
	}

	public void likeSi(String 前缀, int 键的关键字, IDoEachT<byte[], byte[]> call) {
		likeBb(MyFun.u8str2Bytes(前缀), MyFun.int2ByteArray(键的关键字), call);
	}

	/////////////////////////////// left ////////////////////////
	/**
	 * 查询包含前缀的数据,并处理 <br>
	 * 已测
	 * 
	 * @param 前缀   在key的左侧如果匹配,执行回调函数
	 * @param call 回调函数
	 */
	public void leftBb(byte[] 前缀, IDoEachT<byte[], byte[]> call) {
		if (call == null)
			return;
		do遍历操作(前缀, (k, i) -> {
			return MyByte.subBytes(k, i);
		}, call);
	}

	/**
	 * 查询包含前缀的数据,并处理<br>
	 * 已测
	 * 
	 * @param 前缀   在key的左侧如果匹配,执行回调函数
	 * @param call 回调函数
	 */
	public void leftSb(String 前缀, IDoEachT<byte[], byte[]> call) {
		leftBb(MyFun.u8str2Bytes(前缀), call);
	}

	/**
	 * 查询包含前缀的数据,并处理<br>
	 * 已测
	 * 
	 * @param 前缀   在key的左侧如果匹配,执行回调函数
	 * @param call 回调函数
	 */
	public void leftSs(String 前缀, IDoEachT<String, byte[]> call) {
		do遍历操作(MyFun.u8str2Bytes(前缀), (k, i) -> {
			byte[] b = MyByte.subBytes(k, i);
			return MyFun.bytes2U8str(b);
		}, call);
	}

	/**
	 * 查询包含前缀的数据,并处理<br>
	 * 已测
	 * 
	 * @param 前缀   在key的左侧如果匹配,执行回调函数
	 * @param call 回调函数
	 */
	public void leftSi(String 前缀, IDoEachT<Integer, byte[]> call) {
		if (call == null)
			return;
		do遍历操作(MyFun.u8str2Bytes(前缀), (k, i) -> {
			byte[] b = MyByte.subBytes(k, i);
			return MyFun.byteArray2Int(b);
		}, call);
	}

	//////////////////////////////////// 以上是map部分//////////////////////////
	/**
	 * 取出key=主题的 数据,并从列表中删除<br>
	 * 已测
	 * 
	 * @param 主题
	 * @return
	 */
	public MyKV<String, byte[]> getDelS(String 键) {
		MyKV<String, byte[]> mp2 = new MyKV<>();
		mp2.set(键, getS(键));
		delS(键);
		return mp2;
	}

	/**
	 * 取出 key=主题的 数据,并从列表中删除<br>
	 * 已测
	 * 
	 * @param 主题
	 * @return
	 */
	public MyKV<byte[], byte[]> getDelB(byte[] 键) {
		MyKV<byte[], byte[]> mp2 = new MyKV<>();
		mp2.set(键, getB(键));
		delB(键);
		return mp2;
	}

	/**
	 * 随机取出一条数据 ,类似于队列出栈 <br>
	 * 已测
	 * 
	 * @param 主题
	 * @return
	 */
	public MyKV<byte[], byte[]> takeSb(String 主题) {
		return take(MyFun.u8str2Bytes(主题), (k, i) -> {
			byte[] b = MyByte.subBytes(k, i);
			return b;
		});
	}

	/**
	 * 随机取出一条数据 ,类似于队列出栈 <br>
	 * 已测
	 * 
	 * @param 主题
	 * @return
	 */
	public MyKV<String, byte[]> takeSs(String 主题) {
		return take(MyFun.u8str2Bytes(主题), (k, i) -> {
			byte[] b = MyByte.subBytes(k, i);
			return MyFun.bytes2U8str(b);
		});
	}

	/**
	 * 随机取出一条数据,并从列表中删除<br>
	 * 已测
	 * 
	 * @param 主题
	 * @return
	 */
	public MyKV<Long, byte[]> takeSt(String 主题) {
		return take(MyFun.u8str2Bytes(主题), (k, i) -> {
			byte[] b = MyByte.subBytes(k, i);
			return MyFun.byteArray2Long(b);
		});
	}

	/////////////////////////////////// 队列///////////////////////////
	/**
	 * 插入一条数据 <br>
	 * 已测
	 * 
	 * @param key   字符串类型
	 * @param value
	 */
	public void push(String 组, byte[] value) {
		byte[] key = concat(MyFun.u8str2Bytes(组), MyFun.long2ByteArray(MyFun.getNs()));
		putB(key, value);
	}

	/**
	 * 已测
	 * 
	 * @param 组
	 * @return
	 */
	public byte[] pull(String 组) {
		MyKV<Long, byte[]> kv = take(MyFun.u8str2Bytes(组), (k, i) -> {
			byte[] b = MyByte.subBytes(k, i);
			return MyFun.byteArray2Long(b);
		});
		return kv.getValue();
	}

	//////////////////////////////////////////////////
	/**
	 * 异步批量拉取数据<br>
	 * 已测
	 * 
	 * @param 主题
	 * @param 条数
	 * @param call key 是long时间戳,value 是byte[]
	 */

	public void take连续Sb(String 主题, IDoEachT<byte[], byte[]> call) {
		do遍历操作后移除(MyFun.u8str2Bytes(主题), (k, i) -> {
			return MyByte.subBytes(k, i);
		}, call);
	}

	/**
	 * 建议用异步方式拉取<br>
	 * 已测
	 * 
	 * @param 主题
	 * @param call
	 */
	public void take连续Ss(String 主题, IDoEachT<String, byte[]> call) {
		do遍历操作后移除(MyFun.u8str2Bytes(主题), (k, i) -> {
			byte[] bb = MyByte.subBytes(k, i);
			return MyFun.bytes2U8str(bb);
		}, call);
	}

	/**
	 * 连续拉取<br>
	 * 已测
	 * 
	 * @param 主题
	 * @param call key 是long时间戳,value 是byte[]
	 */
	public void take连续SL(String 主题, IDoEachT<Long, byte[]> call) {

		do遍历操作后移除(MyFun.u8str2Bytes(主题), (k, i) -> {
			return MyFun.byteArray2Long(k, i);
		}, call);
	}

	/**
	 * 拉取所有 <br>
	 * 已测
	 * 
	 * @param 组
	 * @return
	 */
	public List<byte[]> takeAll(String 组) {
		List<byte[]> list = new ArrayList<>();
		IDoEachT<byte[], byte[]> call = new IDoEachT<byte[], byte[]>() {
			@Override
			public boolean isOver(byte[] key, byte[] val) throws Exception {
				list.add(val);
				return false;
			}
		};
		do遍历操作后移除(MyFun.u8str2Bytes(组), (k, i) -> {
			return MyByte.subBytes(k, i);
		}, call);
		return list;
	}

	/**
	 * 随机取出一条不做处理,并从库中删除 <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @return
	 */
	public boolean removeTop(byte[] 前缀) {
		DBIterator iterator = getIterator(前缀);
		while (iterator.hasNext()) {

			Map.Entry<byte[], byte[]> mp = iterator.next();
			if (MyByte.leftLike(mp.getKey(), 前缀)) {
				delB(mp.getKey());
				return true;
			}
		}
		return false;

	}

//////////////////////////////////////////////////公共函数+接口//////////////////////////////////
	interface ITransform<T> {
		T s2t(byte[] m, int offset);
	}

	/**
	 * 取出一条满足条件的数据,并从库中删除该数据 <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param 把kv库中的键转换成T类型
	 * @return
	 */
	public <T> MyKV<T, byte[]> take(byte[] 前缀, ITransform<T> trans) {

		while (true) {
			DBIterator iterator = getIterator(前缀);
			MyKV<T, byte[]> mp2 = new MyKV<>();
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();
				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					// byte[] b = ByteUtils.subBytes(mp.getKey(), 前缀.length);
					T t = trans.s2t(mp.getKey(), MyFun.get字节数(前缀));
					mp2.setKey(t);
					mp2.setValue(mp.getValue());
					delB(mp.getKey());
					return mp2;
				}
			}
			// MyDate.sleep(10);
		}

	}

	/**
	 * * <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param trans 把kv库中的键转换成T类型
	 * @param call
	 */
	public <T> void do遍历操作后移除(byte[] 前缀, ITransform<T> trans, IDoEachT<T, byte[]> call) {
		if (call == null || trans == null)
			return;
		DBIterator iterator = getIterator(前缀);
		WriteBatch batch = _db.createWriteBatch();
		try {
			boolean exit = true;
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();
//				if (mp == null)
//					continue;
				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					try {
						T t = trans.s2t(mp.getKey(), MyFun.get字节数(前缀));
						exit = call.isOver(t, mp.getValue());
						if (exit)
							break;
					} catch (Exception e) {
						e.printStackTrace();
						break;
					}
					batch.delete(mp.getKey());
				}
			}
			_db.write(batch);
		} finally {
			try {
				iterator.close();
				batch.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param call
	 */
	public void do遍历操作(byte[] 前缀, IDoEachT<byte[], byte[]> call) {
		if (call == null)
			return;
		DBIterator iterator = getIterator(前缀);
		try {
			boolean exit = true;
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();

				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					try {
						exit = call.isOver(mp.getKey(), mp.getValue());
						// 返回true 退出遍历
						if (exit)
							break;
					} catch (Exception e) {
						e.printStackTrace();
						break;
					}
				}

			}
		} finally {
			try {
				iterator.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public void do遍历键操作(byte[] 前缀, IFilter<byte[]> call) {
		if (call == null)
			return;
		DBIterator iterator = getIterator(前缀);
		try {
			boolean exit = true;
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();

				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					try {
						exit = call.where(mp.getKey());
						if (exit)// 返回true 退出遍历
							break;
					} catch (Exception e) {
						e.printStackTrace();
						break;
					}
				}
			}
		} finally {
			try {
				iterator.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param trans 把kv库中的键转换成T类型
	 * @param call  找到以后的回调函数
	 */
	public <T> void do遍历操作(byte[] 前缀, ITransform<T> 键转换器, IDoEachT<T, byte[]> call) {
		if (call == null)
			return;
		DBIterator iterator = getIterator(前缀);
		try {
			boolean exit = true;
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();

				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					try {
						T key = 键转换器.s2t(mp.getKey(), MyFun.get字节数(前缀));
						exit = call.isOver(key, mp.getValue());
						if (exit)
							break;
					} catch (Exception e) {
						e.printStackTrace();
						break;
					}
				}

			}
		} finally {
			try {
				iterator.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param call
	 */
	public void do遍历后批操作带移除(byte[] 前缀, IDoMapT<WriteBatch, Map.Entry<byte[], byte[]>> call) {
		if (call == null)
			return;

		DBIterator iterator = getIterator(前缀); // seek 有时不起作用
		WriteBatch batch = _db.createWriteBatch();
		try {
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();

				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					try {
						call.run(batch, mp);
						batch.delete(mp.getKey());
					} catch (Exception e) {
						e.printStackTrace();
						break;
					}
				}
			}
			_db.write(batch);
		} finally {
			try {
				iterator.close();
				batch.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param call 遍历到以后 回调执行批指令
	 */
	public void do遍历后批操作(IDoMapT<WriteBatch, Map.Entry<byte[], byte[]>> call) {
		do遍历后批操作(null, call);
	}

	/**
	 * <br>
	 * 已测
	 * 
	 * @param 前缀
	 * @param call
	 */
	public void do遍历后批操作(byte[] 前缀, IDoMapT<WriteBatch, Map.Entry<byte[], byte[]>> call) {
		if (call == null)
			return;

		DBIterator iterator = getIterator(前缀); // seek 有时不起作用
		WriteBatch batch = _db.createWriteBatch();
		try {
			while (iterator.hasNext()) {
				Map.Entry<byte[], byte[]> mp = iterator.next();

				// System.out.println(new String(mp.getKey()));
				// System.out.println(new String(前缀));
				if (前缀 == null || 前缀.length == 0 || MyByte.leftLike(mp.getKey(), 前缀)) {
					try {
						call.run(batch, mp);// 如果返回终止 也不操作吗?
					} catch (Exception e) {
						e.printStackTrace();
						break;
					}
				}
			}
			_db.write(batch);
		} finally {
			try {
				iterator.close();
				batch.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	/**
	 * 批量插入 <br>
	 * 已测
	 * 
	 * @param call 插入脚本用接口方式注入
	 */

	public boolean do批量操作(IDo1<WriteBatch> call) {
		if (call == null)
			return false;
		WriteBatch batch = _db.createWriteBatch();
		try {
			call.run(batch);
			_db.write(batch);
		} finally {
			try {
				batch.close();
			} catch (IOException e) {
				e.printStackTrace();
				return false;
			}
		}
		return true;

	}

/////////////////////////////////////以上是队列部分///////////////////////////////////////

}
